package com.lzq.mq.common.remote;

import com.lzq.mq.common.cache.BrokerServerSyncFutureManage;
import com.lzq.mq.common.cache.NameServerSyncFutureManager;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.coder.TcpMsgDecoder;
import com.lzq.mq.common.coder.TcpMsgEncoder;
import com.lzq.mq.common.constants.TcpConstants;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;

import java.util.concurrent.TimeUnit;

/**
 * 对Broker进行远程连接的客户端(异步转同步)
 * todo 通过面向对象优化代码结构
 */
public class BrokerNettyRemoteClient {

    private String ip;

    private Integer port;

    public BrokerNettyRemoteClient(String ip, Integer port) {
        this.ip = ip;
        this.port = port;
    }

    private EventLoopGroup clientGroup = new NioEventLoopGroup();
    private Bootstrap bootstrap = new Bootstrap();
    private Channel channel;

    /**
     * 远程连接的初始化
     */
    public void buildConnection() {
        bootstrap.group(clientGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ByteBuf delimiter = Unpooled.copiedBuffer(TcpConstants.DEFAULT_DECODE_CHAR.getBytes());
                ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024 * 8, delimiter));
                ch.pipeline().addLast(new TcpMsgDecoder());
                ch.pipeline().addLast(new TcpMsgEncoder());
                ch.pipeline().addLast(new BrokerRemoteRespHandler());
            }
        });
        ChannelFuture channelFuture = null;
        try {
            channelFuture = bootstrap.connect(ip, port).sync().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (!channelFuture.isSuccess()) {
                        throw new RuntimeException("连接Broker出错!");
                    }
                }
            });
            //初始化建立长链接
            channel = channelFuture.channel();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 发送同步消息
     * @param tcpMsg 消息体
     * @param msgId 消息id
     * @return 接收的消息体
     */
    public TcpMsg sendSyncMsg(TcpMsg tcpMsg, String msgId) {
        channel.writeAndFlush(tcpMsg);
        SyncFuture syncFuture = new SyncFuture();
        syncFuture.setMsgId(msgId);
        BrokerServerSyncFutureManage.put(msgId, syncFuture);
        try {
            return (TcpMsg) syncFuture.get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public TcpMsg sendSyncMsg(TcpMsg tcpMsg, String msgId, long time, TimeUnit timeUnit) {
        channel.writeAndFlush(tcpMsg);
        SyncFuture syncFuture = new SyncFuture();
        syncFuture.setMsgId(msgId);
        BrokerServerSyncFutureManage.put(msgId, syncFuture);
        try {
            return (TcpMsg) syncFuture.get(time, timeUnit);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void sendAsyncMsg(TcpMsg tcpMsg) {
        channel.writeAndFlush(tcpMsg);
    }
}
